package liming.tool.thread;

import liming.tool.file.FileRW;
import liming.tool.lock.CodeCondition;
import liming.tool.lock.CodeLock;
import liming.tool.lock.LockAtomicBoolean;
import liming.tool.lock.LockAtomicLong;
import liming.tool.log.LogPool;
import liming.tool.pool.*;
import liming.tool.runnable.HandRunnable;
import liming.tool.runnable.HaveReturnValueInterface;
import liming.tool.runnable.NoReturnValueInterface;

import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * 可拓展线程池
 */
public class ThreadExecutor implements Serializable {
    /**
     * 线程池配置
     */
    private final Config config;
    /**
     * 日志配置
     */
    private final LogConfig logConfig;
    /**
     * 核心线程池
     */
    private final Core core;

    public ThreadExecutor(String poolName, int maxThreads, int minCoreThreads,
                          int maxCoreThreads, long keepAliveTime, double otherMultiplier,
                          int queueCapacity) {
        this.config = new Config(poolName, maxThreads, minCoreThreads, maxCoreThreads,
                keepAliveTime, otherMultiplier, queueCapacity);
        this.logConfig = new LogConfig(config);

        initBefore(logConfig, config);
        logConfig.lock();
        config.lock();
        this.core = new Core(config, logConfig);
    }



    protected void initBefore(LogConfig logConfig, Config config) {

    }

    // Public API -------------------------------------------------------------
    public CompletableFuture<Void> submit(NoReturnValueInterface task) {
        return core.submit(task);
    }

    public <T> CompletableFuture<T> submit(HaveReturnValueInterface<T> task) {
        return core.submit(task);
    }

    public void shutdown() {
        core.shutdown();
    }

    public void awaitAvailability() {
        core.awaitAvailability();
    }

    public List<LogPool.LogMessage> getLogs() {
        return LogPool.getLogs(new LogPool.GetName().add(config.poolName));
    }


    protected void log(Object message) {
        logConfig.log(message);
    }

    public void restart() {
        core.restart();
    }

    public boolean isShutdown() {
        return core.isShutdown();
    }

    // Configuration Classes ---------------------------------------------------

    /**
     * 线程池配置
     */
    public static class Config {
        private final String poolName; // 线程池名称
        private final String uniqueId; // 线程池唯一标识
        private int maxThreads; // 最大线程数
        private int minCoreThreads; // 最小核心线程数
        private int maxCoreThreads; // 最大核心线程数
        private long keepAliveTime; // 线程池中核心线程的存活时间
        private double otherMultiplier; // 其他线程池中线程的存活时间
        private int queueCapacity; // 队列容量

        private boolean isFinal = false;

        /**
         * 构造线程池配置
         *
         * @param poolName        线程池名称
         * @param maxThreads      最大线程数
         * @param minCoreThreads  最小核心线程数
         * @param maxCoreThreads  最大核心线程数
         * @param keepAliveTime   线程池中核心线程的存活时间
         * @param otherMultiplier 额外线程池中线程的存活时间
         * @param queueCapacity   队列容量
         */
        Config(String poolName, int maxThreads, int minCoreThreads,
               int maxCoreThreads, long keepAliveTime, double otherMultiplier,
               int queueCapacity) {
            this.poolName = poolName == null || poolName.isEmpty() ? "default-pool" : poolName;
            this.uniqueId = generateUniqueId(this.poolName);
            this.maxThreads = maxThreads;
            this.minCoreThreads = minCoreThreads;
            this.maxCoreThreads = maxCoreThreads;
            this.keepAliveTime = keepAliveTime;
            this.otherMultiplier = otherMultiplier;
            this.queueCapacity = queueCapacity;
        }

        public String getPoolName() {
            return poolName;
        }

        public String getUniqueId() {
            return uniqueId;
        }

        public int getMaxThreads() {
            return maxThreads;
        }

        public int getMinCoreThreads() {
            return minCoreThreads;
        }

        public int getMaxCoreThreads() {
            return maxCoreThreads;
        }

        public long getKeepAliveTime() {
            return keepAliveTime;
        }

        public double getOtherMultiplier() {
            return otherMultiplier;
        }

        public int getQueueCapacity() {
            return queueCapacity;
        }

        protected final void setMaxThreads(int maxThreads) {
            if (!isFinal)
                this.maxThreads = maxThreads;
        }

        protected void setMinCoreThreads(int minCoreThreads) {
            if (!isFinal)
                this.minCoreThreads = minCoreThreads;
        }

        protected void setMaxCoreThreads(int maxCoreThreads) {
            if (!isFinal)
                this.maxCoreThreads = maxCoreThreads;
        }

        protected void setKeepAliveTime(long keepAliveTime) {
            if (!isFinal)
                this.keepAliveTime = keepAliveTime;
        }

        protected void setOtherMultiplier(double otherMultiplier) {
            if (!isFinal)
                this.otherMultiplier = otherMultiplier;
        }

        protected void setQueueCapacity(int queueCapacity) {
            if (!isFinal)
                this.queueCapacity = queueCapacity;
        }

        /**
         * 验证最大线程数
         */
        private int validateMaxThreads(int value) {
            return Math.max(value, 0);
        }

        private int validateMinCoreThreads(int value) {
            return Math.max(value, 0);
        }

        private int validateMaxCoreThreads(int value) {
            return Math.max(value, minCoreThreads);
        }

        private void lock() {
            isFinal = true;
            this.maxThreads = validateMaxThreads(maxThreads);
            this.minCoreThreads = validateMinCoreThreads(minCoreThreads);
            this.maxCoreThreads = validateMaxCoreThreads(maxCoreThreads);
            this.keepAliveTime = Math.max(0, keepAliveTime);
            this.otherMultiplier = Math.max(0, otherMultiplier);
            this.queueCapacity = Math.max(0, queueCapacity);
        }
    }

    public static class LogConfig {
        private final Config config;// 线程池配置

        private boolean isFinal;

        private File logFile = new File("./logs/executor.log");
        private boolean append = true;
        private boolean consoleLog = false;
        private FileRW.FileEncoding fileEncoding = FileRW.FileEncoding.AUTO;
        private boolean fileLog = false;
        private LogPool.LogFormat fileFormat = LogPool.LogMessage.getDefaultFormat();
        private LogPool.LogFormat consoleFormat = LogPool.LogMessage.getDefaultFormat();

        LogConfig(Config config) {
            this.config = config;
        }

        void log(Object message) {
            log(message, LogPool.getStackTraceElement());
        }

        void log(Object message, StackTraceElement ste) {
            LogPool.LogMessage logMsg = LogPool.logName(ste, config.poolName, message);
            if (consoleLog) {
                System.out.println(consoleFormat.format(logMsg));
            }
            if (fileLog) {
                HandRunnable.run(() -> FileRW.writeTextFile(logFile, fileFormat.format(logMsg) + "\n", append, fileEncoding));
            }
        }

        /**
         * 设置文件日志
         * @param file   文件
         * @param append 是否追加
         */
        public LogConfig enableFileLogging(File file, boolean append, FileRW.FileEncoding fileEncoding) {
            if(isFinal) return this;
            this.logFile = file;
            this.append = append;
            this.fileEncoding = fileEncoding;
            this.fileLog = true;
            return this;
        }

        /**
         * 清空文件日志
         */
        public LogConfig clearLogFile() {
            if(isFinal) return this;
            HandRunnable.run(() -> FileRW.writeTextFile(logFile, "", false));
            return this;
        }

        public LogConfig setShow(boolean showConsoleLog,boolean writeFileLog) {
            if(isFinal) return this;
            this.consoleLog = showConsoleLog;
            this.fileLog = writeFileLog;
            return this;
        }

        /**
         * 锁定配置
         */
        private void lock() {
            isFinal = true;
        }
    }

    // Core Implementation -----------------------------------------------------
    private static class Core {
        /**
         * 线程池配置
         */
        private final Config config;
        /**
         * 日志配置
         */
        private final LogConfig logger;
        /**
         * 任务队列
         */
        private final ObjectsPool<TaskWrapper<?>> taskQueue;
        /**
         * 缓冲队列
         */
        private final ObjectsPool<TaskWrapper<?>> bufferQueue;
        /**
         * 核心线程组
         */
        private final ThreadGroup coreGroup;
        /**
         * 其他线程组
         */
        private final ThreadGroup otherGroup;
        /**
         * 锁
         */
        private final CodeLock lock = new CodeLock();
        /**
         * 线程数量
         */
        private final LockAtomicLong coreThreads = new LockAtomicLong(lock);
        /**
         * 激活线程数量
         */
        private final LockAtomicLong activeCoreThreads = new LockAtomicLong(lock);
        /**
         * 其他线程数量
         */
        private final LockAtomicLong otherThreads = new LockAtomicLong(lock);
        /**
         * 激活其他线程数量
         */
        private final LockAtomicLong activeOtherThreads = new LockAtomicLong(lock);
        private final LockAtomicBoolean shutdown = new LockAtomicBoolean(lock);


        Core(Config config, LogConfig logger) {
            this.config = config;
            this.logger = logger;
            this.taskQueue = new ObjectsPool<>(new CodeLock(), new ArrayList<>(), config.queueCapacity, PoolPolicy.PutPolicy.BLOCK, PoolPolicy.TakePolicy.BLOCK);
            this.bufferQueue = new ObjectsPool<>(new CodeLock(), new ArrayList<>(), config.maxThreads, PoolPolicy.PutPolicy.BLOCK, PoolPolicy.TakePolicy.BLOCK);
            this.coreGroup = new ThreadGroup(config.uniqueId + "-core");
            this.otherGroup = new ThreadGroup(config.uniqueId + "-other");
            startDispatcher();

            startHousekeeper();
        }

        /**
         * 启动调度器
         */
        private void startDispatcher() {
            new Thread(() -> {
                logger.log("调度 启动");
                while (!taskQueue.isClosed()) {
                    HandRunnable.run(new NoReturnValueInterface() {
                        @Override
                        public void run() throws Exception {
                            ObjectsPool<TaskWrapper<?>>.Result take = taskQueue.take();
                            if (Objects.requireNonNull(take.status()) == ObjectsPool.Status.SUCCESS) {
                                ObjectsPool<TaskWrapper<?>>.Result put = bufferQueue.put(take.object());
                                if (put.status() == ObjectsPool.Status.SUCCESS)
                                    adjustThreadPool();
                            }
                        }

                        @Override
                        public void onException(Exception e) {
                            logger.log("调度 异常: " + e.getMessage());
                        }
                    });
                }
                logger.log("调度 结束");
            }, config.uniqueId + "-dispatcher").start();
        }

        /**
         * 启动线程池监控
         */
        private void startHousekeeper() {
            new Thread(coreGroup, () -> {
                logger.log("监控 启动");
                while (!taskQueue.isClosed()) {
                    try {
                        monitorThreadUsage();
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
                logger.log("监控 停止");
            }, config.poolName + "-housekeeper").start();
        }

        /**
         * 监控线程池
         */
        private void monitorThreadUsage() {
            logger.log(String.format(
                    "Thread stats - Core: %d/%d (active/total), Other: %d/%d ,take: %d/%d (get/put)  %d  buffer: %d/%d %d",
                    activeCoreThreads.get(), coreThreads.get(),
                    activeOtherThreads.get(), otherThreads.get()
                    , taskQueue.getTakeCount(), taskQueue.getPutCount(), taskQueue.getNowSize()
                    , bufferQueue.getTakeCount(), bufferQueue.getPutCount(), bufferQueue.getNowSize()
            ), LogPool.getStackTraceElement());
        }

        /**
         * 调整线程池
         */
        private void adjustThreadPool() {
            if(coreThreads.get() < config.maxCoreThreads){
                if(coreThreads.get() < config.minCoreThreads)
                    createCoreThread();
                else if(activeCoreThreads.get() >= coreThreads.get())
                    createCoreThread();

            }else if (config.maxThreads == 0 ||
                    (coreThreads.get() + otherThreads.get() < config.maxThreads))
                createOtherThread();
            else logger.log("调整线程池 - 无法创建线程 "+(coreThreads.get()+otherThreads.get())+"/"+config.maxThreads+"("+config.minCoreThreads+" "+config.maxCoreThreads+")");
        }

        private void createCoreThread() {
            coreThreads.add(1);
            new Thread(coreGroup, new Worker(true),
                    config.poolName + "-core-" + coreThreads.get()).start();
        }

        private void createOtherThread() {
            otherThreads.add(1);
            new Thread(otherGroup, new Worker(false),
                    config.poolName + "-other-" + otherThreads.get()).start();
        }

        public <T> CompletableFuture<T> submit(HaveReturnValueInterface<T> task) {
            TaskWrapper<T> wrapper = new TaskWrapper<>(task);
            taskQueue.put(wrapper);
            return wrapper.future;
        }

        public CompletableFuture<Void> submit(NoReturnValueInterface task) {
            TaskWrapper<Void> wrapper = new TaskWrapper<>(task);
            taskQueue.put(wrapper);
            return wrapper.future;
        }

        public void shutdown() {
            monitorThreadUsage();
            taskQueue.close();
            bufferQueue.close();
            shutdown.set(true);
            monitorThreadUsage();
        }
        public boolean isShutdown() {
            return shutdown.get();
        }

        // 新增重启方法
        public void restart() {
            if (!shutdown.get()) return;
            this.taskQueue.reopen();
            this.bufferQueue.reopen();
            shutdown.set(false);
            startDispatcher();
            startHousekeeper();
        }

        public void awaitAvailability() {
            logger.log("等待线程池空闲");
            CodeCondition condition = lock.newCondition();
            PoolEventCenter.registerEvent(config.uniqueId+"-take", taskQueue);
            PoolEventCenter.registerEvent(config.uniqueId+"-buff", bufferQueue);
            PoolEventInterface<TaskWrapper<?>> poolEventInterface = (event, pool, value) -> condition.signal();
            PoolEventCenter.addEvent(taskQueue, poolEventInterface,PoolEventEnum.empty);
            PoolEventCenter.addEvent(bufferQueue, poolEventInterface,PoolEventEnum.empty);
            while (!taskQueue.isEmpty() || !bufferQueue.isEmpty() || activeOtherThreads.get() != 0) {
                monitorThreadUsage();
                switch (condition.await(config.keepAliveTime, TimeUnit.MILLISECONDS)) {
                    case INTERRUPT:
                        logger.log("Worker terminated: " + Thread.currentThread().getName());
                        break;
                    case TIMEOUT:
                        logger.log("Worker timed out: " + Thread.currentThread().getName());
                        break;
                    case NORMAL:
                        break;
                }
            }
            PoolEventCenter.unRegisterEvent(config.uniqueId+"-take", taskQueue);
            PoolEventCenter.unRegisterEvent(config.uniqueId+"-buff", bufferQueue);
            monitorThreadUsage();
            logger.log("线程池空闲");
        }

        // Worker Thread Implementation ----------------------------------------
        private class Worker implements Runnable {
            private final boolean isCore;
            private final long keepAliveTime;

            Worker(boolean isCore) {
                this.isCore = isCore;
                this.keepAliveTime = isCore ? config.keepAliveTime :
                        (long) (config.keepAliveTime * config.otherMultiplier);
            }

            @Override
            public void run() {
                try {
                    logger.log("Worker started: " + Thread.currentThread().getName());
                    LockAtomicBoolean running = new LockAtomicBoolean(new CodeLock());
                    running.set(true);
                    while (!bufferQueue.isClosed()&&running.get()) {
                        HandRunnable.run(() -> {
                            ObjectsPool<TaskWrapper<?>>.Result task = bufferQueue.take(keepAliveTime, TimeUnit.MILLISECONDS);
                            if (task.status() != ObjectsPool.Status.SUCCESS && shouldTerminate()) {
                                running.set(false);
                                return;
                            }
                            if(task.object()==null) return;
                            updateActiveCount(1);
                            try {
                                task.object().run();
                            } finally {
                                updateActiveCount(-1);
                            }
                        });
                    }
                } finally {
                    updateThreadCount();
                    logger.log("Worker terminated: " + Thread.currentThread().getName());
                }
            }

            private boolean shouldTerminate() {
                return !isCore || coreThreads.get() > config.minCoreThreads;
            }

            private void updateActiveCount(int delta) {
                if (isCore) {
                    activeCoreThreads.add(delta);
                } else {
                    activeOtherThreads.add(delta);
                }
            }

            private void updateThreadCount() {
                if (isCore) {
                    coreThreads.add(-1);
                } else {
                    otherThreads.add(-1);
                }
            }
        }
    }

    // Task Wrapper ------------------------------------------------------------

    /**
     * @param <T>
     */
    private static class TaskWrapper<T> implements Runnable {
        final CompletableFuture<T> future = new CompletableFuture<>();
        private final Supplier<T> task;

        TaskWrapper(HaveReturnValueInterface<T> task) {
            this.task = () -> {
                try {
                    return task.run();
                } catch (Exception e) {
                    return task.onException(e);
                }
            };
        }

        TaskWrapper(NoReturnValueInterface task) {
            this.task = () -> {
                try {
                    task.run();
                    return null;
                } catch (Exception e) {
                    task.onException(e);
                    return null;
                }
            };
        }

        @Override
        public void run() {
            try {
                T result = task.get();
                future.complete(result);
            } catch (Exception e) {
                future.completeExceptionally(e);
            }
        }
    }

    // Utility Methods ---------------------------------------------------------
    private static String generateUniqueId(String poolName) {
        return poolName + "-" + System.currentTimeMillis() + "-" +
                Objects.hash(Thread.currentThread(), System.nanoTime());
    }
}